/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.bff.gaia.mix.api.datastream;

import com.bff.gaia.annotation.Internal;
import com.bff.gaia.annotation.Public;
import com.bff.gaia.annotation.PublicEvolving;
import com.bff.gaia.api.common.ExecutionConfig;
import com.bff.gaia.api.common.functions.FilterFunction;
import com.bff.gaia.api.common.functions.FlatMapFunction;
import com.bff.gaia.api.common.functions.MapFunction;
import com.bff.gaia.api.common.functions.Partitioner;
import com.bff.gaia.api.common.functions.RichFilterFunction;
import com.bff.gaia.api.common.functions.RichFlatMapFunction;
import com.bff.gaia.api.common.functions.RichMapFunction;
import com.bff.gaia.api.common.io.OutputFormat;
import com.bff.gaia.api.common.operators.Keys;
import com.bff.gaia.api.common.operators.ResourceSpec;
import com.bff.gaia.api.common.serialization.SerializationSchema;
import com.bff.gaia.api.common.state.MapStateDescriptor;
import com.bff.gaia.api.common.typeinfo.BasicArrayTypeInfo;
import com.bff.gaia.api.common.typeinfo.PrimitiveArrayTypeInfo;
import com.bff.gaia.api.common.typeinfo.TypeInformation;
import com.bff.gaia.api.java.Utils;
import com.bff.gaia.api.java.functions.KeySelector;
import com.bff.gaia.api.java.io.CsvOutputFormat;
import com.bff.gaia.api.java.io.TextOutputFormat;
import com.bff.gaia.api.java.tuple.Tuple;
import com.bff.gaia.api.java.typeutils.InputTypeConfigurable;
import com.bff.gaia.api.java.typeutils.TypeExtractor;
import com.bff.gaia.core.fs.FileSystem.WriteMode;
import com.bff.gaia.core.fs.Path;
import com.bff.gaia.mix.api.TimeCharacteristic;
import com.bff.gaia.mix.api.collector.selector.OutputSelector;
import com.bff.gaia.mix.api.environment.MixStreamExecutionEnvironment;
import com.bff.gaia.mix.api.functions.AssignerWithPeriodicWatermarks;
import com.bff.gaia.mix.api.functions.AssignerWithPunctuatedWatermarks;
import com.bff.gaia.mix.api.functions.ProcessFunction;
import com.bff.gaia.mix.api.functions.TimestampExtractor;
import com.bff.gaia.mix.api.functions.sink.OutputFormatSinkFunction;
import com.bff.gaia.mix.api.functions.sink.PrintSinkFunction;
import com.bff.gaia.mix.api.functions.sink.SinkFunction;
import com.bff.gaia.mix.api.functions.sink.SocketClientSink;
import com.bff.gaia.mix.api.functions.timestamps.AscendingTimestampExtractor;
import com.bff.gaia.mix.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import com.bff.gaia.mix.api.operators.OneInputMixStreamOperator;
import com.bff.gaia.mix.api.operators.ProcessOperator;
import com.bff.gaia.mix.api.operators.MixStreamFilter;
import com.bff.gaia.mix.api.operators.MixStreamFlatMap;
import com.bff.gaia.mix.api.operators.MixStreamMap;
import com.bff.gaia.mix.api.operators.MixStreamSink;
import com.bff.gaia.mix.api.transformations.OneInputTransformation;
import com.bff.gaia.mix.api.transformations.PartitionTransformation;
import com.bff.gaia.mix.api.transformations.MixStreamTransformation;
import com.bff.gaia.mix.api.transformations.UnionTransformation;
import com.bff.gaia.mix.api.windowing.assigners.GlobalWindows;
import com.bff.gaia.mix.api.windowing.assigners.SlidingEventTimeWindows;
import com.bff.gaia.mix.api.windowing.assigners.SlidingProcessingTimeWindows;
import com.bff.gaia.mix.api.windowing.assigners.TumblingEventTimeWindows;
import com.bff.gaia.mix.api.windowing.assigners.TumblingProcessingTimeWindows;
import com.bff.gaia.mix.api.windowing.assigners.WindowAssigner;
import com.bff.gaia.mix.api.windowing.evictors.CountEvictor;
import com.bff.gaia.mix.api.windowing.time.Time;
import com.bff.gaia.mix.api.windowing.triggers.CountTrigger;
import com.bff.gaia.mix.api.windowing.triggers.PurgingTrigger;
import com.bff.gaia.mix.api.windowing.windows.GlobalWindow;
import com.bff.gaia.mix.api.windowing.windows.TimeWindow;
import com.bff.gaia.mix.api.windowing.windows.Window;
import com.bff.gaia.mix.api.xjoin.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin.operators.XjoinMixStreamV3;
import com.bff.gaia.mix.api.xjoin2.datastream.XjoinMixStreamV4;
import com.bff.gaia.mix.runtime.operators.ExtractTimestampsOperator;
import com.bff.gaia.mix.runtime.operators.TimestampsAndPeriodicWatermarksOperator;
import com.bff.gaia.mix.runtime.operators.TimestampsAndPunctuatedWatermarksOperator;
import com.bff.gaia.mix.runtime.partitioner.BroadcastPartitioner;
import com.bff.gaia.mix.runtime.partitioner.CustomPartitionerWrapper;
import com.bff.gaia.mix.runtime.partitioner.ForwardPartitioner;
import com.bff.gaia.mix.runtime.partitioner.GlobalPartitioner;
import com.bff.gaia.mix.runtime.partitioner.RebalancePartitioner;
import com.bff.gaia.mix.runtime.partitioner.RescalePartitioner;
import com.bff.gaia.mix.runtime.partitioner.ShufflePartitioner;
import com.bff.gaia.mix.runtime.partitioner.MixStreamPartitioner;
import com.bff.gaia.mix.util.keys.KeySelectorUtil;
import com.bff.gaia.util.Preconditions;

import java.util.ArrayList;
import java.util.List;

/**
 * A DataMixStream represents a stream of elements of the same type. A DataMixStream
 * can be transformed into another DataMixStream by applying a transformation as
 * for example:
 * <ul>
 * <li>{@link DataMixStream#map}
 * <li>{@link DataMixStream#filter}
 * </ul>
 *
 * @param <T> The type of the elements in this stream.
 */
@Public
public class DataMixStream<T> {

	protected final MixStreamExecutionEnvironment environment;

	protected final MixStreamTransformation<T> transformation;

	/**
	 * Create a new {@link DataMixStream} in the given execution environment with
	 * partitioning set to forward by default.
	 *
	 * @param environment The MixStreamExecutionEnvironment
	 */
	public DataMixStream(MixStreamExecutionEnvironment environment, MixStreamTransformation<T> transformation) {
		this.environment = Preconditions.checkNotNull(environment, "Execution Environment must not be null.");
		this.transformation = Preconditions.checkNotNull(transformation, "Stream Transformation must not be null.");
	}

	/**
	 * Returns the ID of the {@link DataMixStream} in the current {@link MixStreamExecutionEnvironment}.
	 *
	 * @return ID of the DataMixStream
	 */
	@Internal
	public int getId() {
		return transformation.getId();
	}

	/**
	 * Gets the parallelism for this operator.
	 *
	 * @return The parallelism set for this operator.
	 */
	public int getParallelism() {
		return transformation.getParallelism();
	}

	/**
	 * Gets the minimum resources for this operator.
	 *
	 * @return The minimum resources set for this operator.
	 */
	@PublicEvolving
	public ResourceSpec getMinResources() {
		return transformation.getMinResources();
	}

	/**
	 * Gets the preferred resources for this operator.
	 *
	 * @return The preferred resources set for this operator.
	 */
	@PublicEvolving
	public ResourceSpec getPreferredResources() {
		return transformation.getPreferredResources();
	}

	/**
	 * Gets the type of the stream.
	 *
	 * @return The type of the datastream.
	 */
	public TypeInformation<T> getType() {
		return transformation.getOutputType();
	}

	/**
	 * Invokes the {@link com.bff.gaia.api.java.ClosureCleaner}
	 * on the given function if closure cleaning is enabled in the {@link ExecutionConfig}.
	 *
	 * @return The cleaned Function
	 */
	protected <F> F clean(F f) {
		return getExecutionEnvironment().clean(f);
	}

	/**
	 * Returns the {@link MixStreamExecutionEnvironment} that was used to create this
	 * {@link DataMixStream}.
	 *
	 * @return The Execution Environment
	 */
	public MixStreamExecutionEnvironment getExecutionEnvironment() {
		return environment;
	}

	public ExecutionConfig getExecutionConfig() {
		return environment.getConfig();
	}

	/**
	 * Creates a new {@link DataMixStream} by merging {@link DataMixStream} outputs of
	 * the same type with each other. The DataMixStreams merged using this operator
	 * will be transformed simultaneously.
	 *
	 * @param streams The DataMixStreams to union output with.
	 * @return The {@link DataMixStream}.
	 */
	@SafeVarargs
	public final DataMixStream<T> union(DataMixStream<T>... streams) {
		List<MixStreamTransformation<T>> unionedTransforms = new ArrayList<>();
		unionedTransforms.add(this.transformation);

		for (DataMixStream<T> newMixStream : streams) {
			if (!getType().equals(newMixStream.getType())) {
				throw new IllegalArgumentException("Cannot union streams of different types: "
					+ getType() + " and " + newMixStream.getType());
			}

			unionedTransforms.add(newMixStream.getTransformation());
		}
		return new DataMixStream<>(this.environment, new UnionTransformation<>(unionedTransforms));
	}

	/**
	 * Operator used for directing tuples to specific named outputs using an
	 * {@link com.bff.gaia.mix.api.collector.selector.OutputSelector}.
	 * Calling this method on an operator creates a new {@link SplitMixStream}.
	 *
	 * @param outputSelector The user defined
	 *                       {@link com.bff.gaia.mix.api.collector.selector.OutputSelector}
	 *                       for directing the tuples.
	 * @return The {@link SplitMixStream}
	 * @deprecated Please use side ouput instead.
	 */
	@Deprecated
	public SplitMixStream<T> split(OutputSelector<T> outputSelector) {
		return new SplitMixStream<>(this, clean(outputSelector));
	}

	/**
	 * Creates a new {@link ConnectedMixStreams} by connecting
	 * {@link DataMixStream} outputs of (possible) different types with each other.
	 * The DataMixStreams connected using this operator can be used with
	 * CoFunctions to apply joint transformations.
	 *
	 * @param dataMixStream The DataMixStream with which this stream will be connected.
	 * @return The {@link ConnectedMixStreams}.
	 */
	public <R> ConnectedMixStreams<T, R> connect(DataMixStream<R> dataMixStream) {
		return new ConnectedMixStreams<>(environment, this, dataMixStream);
	}

	/**
	 * Creates a new {@link BroadcastConnectedMixStream} by connecting the current
	 * {@link DataMixStream} or {@link KeyedMixStream} with a {@link BroadcastMixStream}.
	 *
	 * <p>The latter can be created using the {@link #broadcast(MapStateDescriptor[])} method.
	 *
	 * <p>The resulting stream can be further processed using the {@code BroadcastConnectedMixStream.process(MyFunction)}
	 * method, where {@code MyFunction} can be either a
	 * {@link com.bff.gaia.mix.api.functions.co.KeyedBroadcastProcessFunction KeyedBroadcastProcessFunction}
	 * or a {@link com.bff.gaia.mix.api.functions.co.BroadcastProcessFunction BroadcastProcessFunction}
	 * depending on the current stream being a {@link KeyedMixStream} or not.
	 *
	 * @param broadcastMixStream The broadcast stream with the broadcast state to be connected with this stream.
	 * @return The {@link BroadcastConnectedMixStream}.
	 */
	@PublicEvolving
	public <R> BroadcastConnectedMixStream<T, R> connect(BroadcastMixStream<R> broadcastMixStream) {
		return new BroadcastConnectedMixStream<>(
			environment,
			this,
			Preconditions.checkNotNull(broadcastMixStream),
			broadcastMixStream.getBroadcastStateDescriptor());
	}

	/**
	 * It creates a new {@link KeyedMixStream} that uses the provided key for partitioning
	 * its operator states.
	 *
	 * @param key The KeySelector to be used for extracting the key for partitioning
	 * @return The {@link DataMixStream} with partitioned state (i.e. KeyedMixStream)
	 */
	public <K> KeyedMixStream<T, K> keyBy(KeySelector<T, K> key) {
		Preconditions.checkNotNull(key);
		return new KeyedMixStream<>(this, clean(key));
	}

	/**
	 * It creates a new {@link KeyedMixStream} that uses the provided key with explicit type information
	 * for partitioning its operator states.
	 *
	 * @param key     The KeySelector to be used for extracting the key for partitioning.
	 * @param keyType The type information describing the key type.
	 * @return The {@link DataMixStream} with partitioned state (i.e. KeyedMixStream)
	 */
	public <K> KeyedMixStream<T, K> keyBy(KeySelector<T, K> key, TypeInformation<K> keyType) {
		Preconditions.checkNotNull(key);
		Preconditions.checkNotNull(keyType);
		return new KeyedMixStream<>(this, clean(key), keyType);
	}

	/**
	 * Partitions the operator state of a {@link DataMixStream} by the given key positions.
	 *
	 * @param fields The position of the fields on which the {@link DataMixStream}
	 *               will be grouped.
	 * @return The {@link DataMixStream} with partitioned state (i.e. KeyedMixStream)
	 */
	public KeyedMixStream<T, Tuple> keyBy(int... fields) {
		if (getType() instanceof BasicArrayTypeInfo || getType() instanceof PrimitiveArrayTypeInfo) {
			return keyBy(KeySelectorUtil.getSelectorForArray(fields, getType()));
		} else {
			return keyBy(new Keys.ExpressionKeys<>(fields, getType()));
		}
	}

	/**
	 * Partitions the operator state of a {@link DataMixStream} using field expressions.
	 * A field expression is either the name of a public field or a getter method with parentheses
	 * of the {@link DataMixStream}'s underlying type. A dot can be used to drill
	 * down into objects, as in {@code "field1.getInnerField2()" }.
	 *
	 * @param fields One or more field expressions on which the state of the {@link DataMixStream} operators will be
	 *               partitioned.
	 * @return The {@link DataMixStream} with partitioned state (i.e. KeyedMixStream)
	 **/
	public KeyedMixStream<T, Tuple> keyBy(String... fields) {
		return keyBy(new Keys.ExpressionKeys<>(fields, getType()));
	}

	private KeyedMixStream<T, Tuple> keyBy(Keys<T> keys) {
		return new KeyedMixStream<>(this, clean(KeySelectorUtil.getSelectorForKeys(keys,
			getType(), getExecutionConfig())));
	}

	/**
	 * Partitions a tuple DataMixStream on the specified key fields using a custom partitioner.
	 * This method takes the key position to partition on, and a partitioner that accepts the key type.
	 *
	 * <p>Note: This method works only on single field keys.
	 *
	 * @param partitioner The partitioner to assign partitions to keys.
	 * @param field       The field index on which the DataMixStream is to partitioned.
	 * @return The partitioned DataMixStream.
	 */
	public <K> DataMixStream<T> partitionCustom(Partitioner<K> partitioner, int field) {
		Keys.ExpressionKeys<T> outExpressionKeys = new Keys.ExpressionKeys<>(new int[]{field}, getType());
		return partitionCustom(partitioner, outExpressionKeys);
	}

	/**
	 * Partitions a POJO DataMixStream on the specified key fields using a custom partitioner.
	 * This method takes the key expression to partition on, and a partitioner that accepts the key type.
	 *
	 * <p>Note: This method works only on single field keys.
	 *
	 * @param partitioner The partitioner to assign partitions to keys.
	 * @param field       The expression for the field on which the DataMixStream is to partitioned.
	 * @return The partitioned DataMixStream.
	 */
	public <K> DataMixStream<T> partitionCustom(Partitioner<K> partitioner, String field) {
		Keys.ExpressionKeys<T> outExpressionKeys = new Keys.ExpressionKeys<>(new String[]{field}, getType());
		return partitionCustom(partitioner, outExpressionKeys);
	}


	/**
	 * Partitions a DataMixStream on the key returned by the selector, using a custom partitioner.
	 * This method takes the key selector to get the key to partition on, and a partitioner that
	 * accepts the key type.
	 *
	 * <p>Note: This method works only on single field keys, i.e. the selector cannot return tuples
	 * of fields.
	 *
	 * @param partitioner The partitioner to assign partitions to keys.
	 * @param keySelector The KeySelector with which the DataMixStream is partitioned.
	 * @return The partitioned DataMixStream.
	 * @see KeySelector
	 */
	public <K> DataMixStream<T> partitionCustom(Partitioner<K> partitioner, KeySelector<T, K> keySelector) {
		return setConnectionType(new CustomPartitionerWrapper<>(clean(partitioner),
			clean(keySelector)));
	}

	//	private helper method for custom partitioning
	private <K> DataMixStream<T> partitionCustom(Partitioner<K> partitioner, Keys<T> keys) {
		KeySelector<T, K> keySelector = KeySelectorUtil.getSelectorForOneKey(keys, partitioner, getType(), getExecutionConfig());

		return setConnectionType(
			new CustomPartitionerWrapper<>(
				clean(partitioner),
				clean(keySelector)));
	}

	/**
	 * Sets the partitioning of the {@link DataMixStream} so that the output elements
	 * are broadcast to every parallel instance of the next operation.
	 *
	 * @return The DataMixStream with broadcast partitioning set.
	 */
	public DataMixStream<T> broadcast() {
		return setConnectionType(new BroadcastPartitioner<T>());
	}

	/**
	 * Sets the partitioning of the {@link DataMixStream} so that the output elements
	 * are broadcasted to every parallel instance of the next operation. In addition,
	 * it implicitly as many {@link com.bff.gaia.api.common.state.BroadcastState broadcast states}
	 * as the specified descriptors which can be used to store the element of the stream.
	 *
	 * @param broadcastStateDescriptors the descriptors of the broadcast states to create.
	 * @return A {@link BroadcastMixStream} which can be used in the {@link #connect(BroadcastMixStream)} to
	 * create a {@link BroadcastConnectedMixStream} for further processing of the elements.
	 */
	@PublicEvolving
	public BroadcastMixStream<T> broadcast(final MapStateDescriptor<?, ?>... broadcastStateDescriptors) {
		Preconditions.checkNotNull(broadcastStateDescriptors);
		final DataMixStream<T> broadcastMixStream = setConnectionType(new BroadcastPartitioner<>());
		return new BroadcastMixStream<>(environment, broadcastMixStream, broadcastStateDescriptors);
	}

	/**
	 * Sets the partitioning of the {@link DataMixStream} so that the output elements
	 * are shuffled uniformly randomly to the next operation.
	 *
	 * @return The DataMixStream with shuffle partitioning set.
	 */
	@PublicEvolving
	public DataMixStream<T> shuffle() {
		return setConnectionType(new ShufflePartitioner<T>());
	}

	/**
	 * Sets the partitioning of the {@link DataMixStream} so that the output elements
	 * are forwarded to the local subtask of the next operation.
	 *
	 * @return The DataMixStream with forward partitioning set.
	 */
	public DataMixStream<T> forward() {
		return setConnectionType(new ForwardPartitioner<T>());
	}

	/**
	 * Sets the partitioning of the {@link DataMixStream} so that the output elements
	 * are distributed evenly to instances of the next operation in a round-robin
	 * fashion.
	 *
	 * @return The DataMixStream with rebalance partitioning set.
	 */
	public DataMixStream<T> rebalance() {
		return setConnectionType(new RebalancePartitioner<T>());
	}

	/**
	 * Sets the partitioning of the {@link DataMixStream} so that the output elements
	 * are distributed evenly to a subset of instances of the next operation in a round-robin
	 * fashion.
	 *
	 * <p>The subset of downstream operations to which the upstream operation sends
	 * elements depends on the degree of parallelism of both the upstream and downstream operation.
	 * For example, if the upstream operation has parallelism 2 and the downstream operation
	 * has parallelism 4, then one upstream operation would distribute elements to two
	 * downstream operations while the other upstream operation would distribute to the other
	 * two downstream operations. If, on the other hand, the downstream operation has parallelism
	 * 2 while the upstream operation has parallelism 4 then two upstream operations will
	 * distribute to one downstream operation while the other two upstream operations will
	 * distribute to the other downstream operations.
	 *
	 * <p>In cases where the different parallelisms are not multiples of each other one or several
	 * downstream operations will have a differing number of inputs from upstream operations.
	 *
	 * @return The DataMixStream with rescale partitioning set.
	 */
	@PublicEvolving
	public DataMixStream<T> rescale() {
		return setConnectionType(new RescalePartitioner<T>());
	}

	/**
	 * Sets the partitioning of the {@link DataMixStream} so that the output values
	 * all go to the first instance of the next processing operator. Use this
	 * setting with care since it might cause a serious performance bottleneck
	 * in the application.
	 *
	 * @return The DataMixStream with shuffle partitioning set.
	 */
	@PublicEvolving
	public DataMixStream<T> global() {
		return setConnectionType(new GlobalPartitioner<T>());
	}

	/**
	 * Initiates an iterative part of the program that feeds back data streams.
	 * The iterative part needs to be closed by calling
	 * {@link IterativeMixStream#closeWith(DataMixStream)}. The transformation of
	 * this IterativeMixStream will be the iteration head. The data stream
	 * given to the {@link IterativeMixStream#closeWith(DataMixStream)} method is
	 * the data stream that will be fed back and used as the input for the
	 * iteration head. The user can also use different feedback type than the
	 * input of the iteration and treat the input and feedback streams as a
	 * {@link ConnectedMixStreams} be calling
	 * {@link IterativeMixStream#withFeedbackType(TypeInformation)}
	 *
	 * <p>A common usage pattern for mix iterations is to use output
	 * splitting to send a part of the closing data stream to the head. Refer to
	 * {@link #split(OutputSelector)} for more information.
	 *
	 * <p>The iteration edge will be partitioned the same way as the first input of
	 * the iteration head unless it is changed in the
	 * {@link IterativeMixStream#closeWith(DataMixStream)} call.
	 *
	 * <p>By default a DataMixStream with iteration will never terminate, but the user
	 * can use the maxWaitTime parameter to set a max waiting time for the
	 * iteration head. If no data received in the set time, the stream
	 * terminates.
	 *
	 * @return The iterative data stream created.
	 */
	@PublicEvolving
	public IterativeMixStream<T> iterate() {
		return new IterativeMixStream<>(this, 0);
	}

	/**
	 * Initiates an iterative part of the program that feeds back data streams.
	 * The iterative part needs to be closed by calling
	 * {@link IterativeMixStream#closeWith(DataMixStream)}. The transformation of
	 * this IterativeMixStream will be the iteration head. The data stream
	 * given to the {@link IterativeMixStream#closeWith(DataMixStream)} method is
	 * the data stream that will be fed back and used as the input for the
	 * iteration head. The user can also use different feedback type than the
	 * input of the iteration and treat the input and feedback streams as a
	 * {@link ConnectedMixStreams} be calling
	 * {@link IterativeMixStream#withFeedbackType(TypeInformation)}
	 *
	 * <p>A common usage pattern for mix iterations is to use output
	 * splitting to send a part of the closing data stream to the head. Refer to
	 * {@link #split(OutputSelector)} for more information.
	 *
	 * <p>The iteration edge will be partitioned the same way as the first input of
	 * the iteration head unless it is changed in the
	 * {@link IterativeMixStream#closeWith(DataMixStream)} call.
	 *
	 * <p>By default a DataMixStream with iteration will never terminate, but the user
	 * can use the maxWaitTime parameter to set a max waiting time for the
	 * iteration head. If no data received in the set time, the stream
	 * terminates.
	 *
	 * @param maxWaitTimeMillis Number of milliseconds to wait between inputs before shutting
	 *                          down
	 * @return The iterative data stream created.
	 */
	@PublicEvolving
	public IterativeMixStream<T> iterate(long maxWaitTimeMillis) {
		return new IterativeMixStream<>(this, maxWaitTimeMillis);
	}

	/**
	 * Applies a Map transformation on a {@link DataMixStream}. The transformation
	 * calls a {@link MapFunction} for each element of the DataMixStream. Each
	 * MapFunction call returns exactly one element. The user can also extend
	 * {@link RichMapFunction} to gain access to other features provided by the
	 * {@link com.bff.gaia.api.common.functions.RichFunction} interface.
	 *
	 * @param mapper The MapFunction that is called for each element of the
	 *               DataMixStream.
	 * @param <R>    output type
	 * @return The transformed {@link DataMixStream}.
	 */
	public <R> SingleOutputMixStreamOperator<R> map(MapFunction<T, R> mapper) {

		TypeInformation<R> outType = TypeExtractor.getMapReturnTypes(clean(mapper), getType(),
			Utils.getCallLocationName(), true);

		return transform("Map", outType, new MixStreamMap<>(clean(mapper)));
	}

	/**
	 * Applies a FlatMap transformation on a {@link DataMixStream}. The
	 * transformation calls a {@link FlatMapFunction} for each element of the
	 * DataMixStream. Each FlatMapFunction call can return any number of elements
	 * including none. The user can also extend {@link RichFlatMapFunction} to
	 * gain access to other features provided by the
	 * {@link com.bff.gaia.api.common.functions.RichFunction} interface.
	 *
	 * @param flatMapper The FlatMapFunction that is called for each element of the
	 *                   DataMixStream
	 * @param <R>        output type
	 * @return The transformed {@link DataMixStream}.
	 */
	public <R> SingleOutputMixStreamOperator<R> flatMap(FlatMapFunction<T, R> flatMapper) {

		TypeInformation<R> outType = TypeExtractor.getFlatMapReturnTypes(clean(flatMapper),
			getType(), Utils.getCallLocationName(), true);

		return transform("Flat Map", outType, new MixStreamFlatMap<>(clean(flatMapper)));

	}

	/**
	 * Applies the given {@link ProcessFunction} on the input stream, thereby
	 * creating a transformed output stream.
	 *
	 * <p>The function will be called for every element in the input streams and can produce zero
	 * or more output elements.
	 *
	 * @param processFunction The {@link ProcessFunction} that is called for each element
	 *                        in the stream.
	 * @param <R>             The type of elements emitted by the {@code ProcessFunction}.
	 * @return The transformed {@link DataMixStream}.
	 */
	@PublicEvolving
	public <R> SingleOutputMixStreamOperator<R> process(ProcessFunction<T, R> processFunction) {

		TypeInformation<R> outType = TypeExtractor.getUnaryOperatorReturnType(
			processFunction,
			ProcessFunction.class,
			0,
			1,
			TypeExtractor.NO_INDEX,
			getType(),
			Utils.getCallLocationName(),
			true);

		return process(processFunction, outType);
	}

	/**
	 * Applies the given {@link ProcessFunction} on the input stream, thereby
	 * creating a transformed output stream.
	 *
	 * <p>The function will be called for every element in the input streams and can produce zero
	 * or more output elements.
	 *
	 * @param processFunction The {@link ProcessFunction} that is called for each element
	 *                        in the stream.
	 * @param outputType      {@link TypeInformation} for the result type of the function.
	 * @param <R>             The type of elements emitted by the {@code ProcessFunction}.
	 * @return The transformed {@link DataMixStream}.
	 */
	@Internal
	public <R> SingleOutputMixStreamOperator<R> process(
		ProcessFunction<T, R> processFunction,
		TypeInformation<R> outputType) {

		ProcessOperator<T, R> operator = new ProcessOperator<>(clean(processFunction));

		return transform("Process", outputType, operator);
	}

	/**
	 * Applies a Filter transformation on a {@link DataMixStream}. The
	 * transformation calls a {@link FilterFunction} for each element of the
	 * DataMixStream and retains only those element for which the function returns
	 * true. Elements for which the function returns false are filtered. The
	 * user can also extend {@link RichFilterFunction} to gain access to other
	 * features provided by the
	 * {@link com.bff.gaia.api.common.functions.RichFunction} interface.
	 *
	 * @param filter The FilterFunction that is called for each element of the
	 *               DataMixStream.
	 * @return The filtered DataMixStream.
	 */
	public SingleOutputMixStreamOperator<T> filter(FilterFunction<T> filter) {
		return transform("Filter", getType(), new MixStreamFilter<>(clean(filter)));

	}

	/**
	 * Initiates a Project transformation on a {@link Tuple} {@link DataMixStream}.<br>
	 * <b>Note: Only Tuple DataMixStreams can be projected.</b>
	 *
	 * <p>The transformation projects each Tuple of the DataSet onto a (sub)set of
	 * fields.
	 *
	 * @param fieldIndexes The field indexes of the input tuples that are retained. The
	 *                     order of fields in the output tuple corresponds to the order
	 *                     of field indexes.
	 * @return The projected DataMixStream
	 * @see Tuple
	 * @see DataMixStream
	 */
	@PublicEvolving
	public <R extends Tuple> SingleOutputMixStreamOperator<R> project(int... fieldIndexes) {
		return new MixStreamProjection<>(this, fieldIndexes).projectTupleX();
	}

	/**
	 * Creates a join operation. See {@link CoGroupedMixStreams} for an example of how the keys
	 * and window can be specified.
	 */
	public <T2> CoGroupedMixStreams<T, T2> coGroup(DataMixStream<T2> otherMixStream) {
		return new CoGroupedMixStreams<>(this, otherMixStream);
	}

	/**
	 * Creates a join operation. See {@link JoinedMixStreams} for an example of how the keys
	 * and window can be specified.
	 */
	public <T2> JoinedMixStreams<T, T2> join(DataMixStream<T2> otherMixStream) {
		return new JoinedMixStreams<>(this, otherMixStream);
	}

	public XjoinMixStreamV3 xjoin(SideTableInfo sideTableInfo) {
		return new XjoinMixStreamV3(this, sideTableInfo);
	}

	public XjoinMixStreamV4 xjoinV4(com.bff.gaia.mix.api.xjoin2.core.side.SideTableInfo sideTableInfo) {
		return new XjoinMixStreamV4(environment, this, sideTableInfo);
	}

	/**
	 * Windows this {@code DataMixStream} into tumbling time windows.
	 *
	 * <p>This is a shortcut for either {@code .window(TumblingEventTimeWindows.of(size))} or
	 * {@code .window(TumblingProcessingTimeWindows.of(size))} depending on the time characteristic
	 * set using
	 *
	 * <p>Note: This operation is inherently non-parallel since all elements have to pass through
	 * the same operator instance.
	 * <p>
	 * {@link com.bff.gaia.mix.api.environment.MixStreamExecutionEnvironment#setMixStreamTimeCharacteristic(com.bff.gaia.mix.api.TimeCharacteristic)}
	 *
	 * @param size The size of the window.
	 */
	public AllWindowedMixStream<T, TimeWindow> timeWindowAll(Time size) {
		if (environment.getMixStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime) {
			return windowAll(TumblingProcessingTimeWindows.of(size));
		} else {
			return windowAll(TumblingEventTimeWindows.of(size));
		}
	}

	/**
	 * Windows this {@code DataMixStream} into sliding time windows.
	 *
	 * <p>This is a shortcut for either {@code .window(SlidingEventTimeWindows.of(size, slide))} or
	 * {@code .window(SlidingProcessingTimeWindows.of(size, slide))} depending on the time characteristic
	 * set using
	 * {@link com.bff.gaia.mix.api.environment.MixStreamExecutionEnvironment#setMixStreamTimeCharacteristic(com.bff.gaia.mix.api.TimeCharacteristic)}
	 *
	 * <p>Note: This operation is inherently non-parallel since all elements have to pass through
	 * the same operator instance.
	 *
	 * @param size The size of the window.
	 */
	public AllWindowedMixStream<T, TimeWindow> timeWindowAll(Time size, Time slide) {
		if (environment.getMixStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime) {
			return windowAll(SlidingProcessingTimeWindows.of(size, slide));
		} else {
			return windowAll(SlidingEventTimeWindows.of(size, slide));
		}
	}

	/**
	 * Windows this {@code DataMixStream} into tumbling count windows.
	 *
	 * <p>Note: This operation is inherently non-parallel since all elements have to pass through
	 * the same operator instance.
	 *
	 * @param size The size of the windows in number of elements.
	 */
	public AllWindowedMixStream<T, GlobalWindow> countWindowAll(long size) {
		return windowAll(GlobalWindows.create()).trigger(PurgingTrigger.of(CountTrigger.of(size)));
	}

	/**
	 * Windows this {@code DataMixStream} into sliding count windows.
	 *
	 * <p>Note: This operation is inherently non-parallel since all elements have to pass through
	 * the same operator instance.
	 *
	 * @param size  The size of the windows in number of elements.
	 * @param slide The slide interval in number of elements.
	 */
	public AllWindowedMixStream<T, GlobalWindow> countWindowAll(long size, long slide) {
		return windowAll(GlobalWindows.create())
			.evictor(CountEvictor.of(size))
			.trigger(CountTrigger.of(slide));
	}

	/**
	 * Windows this data stream to a {@code KeyedTriggerWindowDataMixStream}, which evaluates windows
	 * over a key grouped stream. Elements are put into windows by a
	 * {@link com.bff.gaia.mix.api.windowing.assigners.WindowAssigner}. The grouping of
	 * elements is done both by key and by window.
	 *
	 * <p>A {@link com.bff.gaia.mix.api.windowing.triggers.Trigger} can be defined to specify
	 * when windows are evaluated. However, {@code WindowAssigners} have a default {@code Trigger}
	 * that is used if a {@code Trigger} is not specified.
	 *
	 * <p>Note: This operation is inherently non-parallel since all elements have to pass through
	 * the same operator instance.
	 *
	 * @param assigner The {@code WindowAssigner} that assigns elements to windows.
	 * @return The trigger windows data stream.
	 */
	@PublicEvolving
	public <W extends Window> AllWindowedMixStream<T, W> windowAll(WindowAssigner<? super T, W> assigner) {
		return new AllWindowedMixStream<>(this, assigner);
	}

	// ------------------------------------------------------------------------
	//  Timestamps and watermarks
	// ------------------------------------------------------------------------

	/**
	 * Extracts a timestamp from an element and assigns it as the internal timestamp of that element.
	 * The internal timestamps are, for example, used to to event-time window operations.
	 *
	 * <p>If you know that the timestamps are strictly increasing you can use an
	 * {@link AscendingTimestampExtractor}. Otherwise,
	 * you should provide a {@link TimestampExtractor} that also implements
	 * {@link TimestampExtractor#getCurrentWatermark()} to keep track of watermarks.
	 *
	 * @param extractor The TimestampExtractor that is called for each element of the DataMixStream.
	 * @see #assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks)
	 * @see #assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks)
	 * @deprecated Please use {@link #assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks)}
	 * of {@link #assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks)}
	 * instead.
	 */
	@Deprecated
	public SingleOutputMixStreamOperator<T> assignTimestamps(TimestampExtractor<T> extractor) {
		// match parallelism to input, otherwise dop=1 sources could lead to some strange
		// behaviour: the watermark will creep along very slowly because the elements
		// from the source go to each extraction operator round robin.
		int inputParallelism = getTransformation().getParallelism();
		ExtractTimestampsOperator<T> operator = new ExtractTimestampsOperator<>(clean(extractor));
		return transform("ExtractTimestamps", getTransformation().getOutputType(), operator)
			.setParallelism(inputParallelism);
	}

	/**
	 * Assigns timestamps to the elements in the data stream and periodically creates
	 * watermarks to signal event time progress.
	 *
	 * <p>This method creates watermarks periodically (for example every second), based
	 * on the watermarks indicated by the given watermark generator. Even when no new elements
	 * in the stream arrive, the given watermark generator will be periodically checked for
	 * new watermarks. The interval in which watermarks are generated is defined in
	 * {@link ExecutionConfig#setAutoWatermarkInterval(long)}.
	 *
	 * <p>Use this method for the common cases, where some characteristic over all elements
	 * should generate the watermarks, or where watermarks are simply trailing behind the
	 * wall clock time by a certain amount.
	 *
	 * <p>For the second case and when the watermarks are required to lag behind the maximum
	 * timestamp seen so far in the elements of the stream by a fixed amount of time, and this
	 * amount is known in advance, use the
	 * {@link BoundedOutOfOrdernessTimestampExtractor}.
	 *
	 * <p>For cases where watermarks should be created in an irregular fashion, for example
	 * based on certain markers that some element carry, use the
	 * {@link AssignerWithPunctuatedWatermarks}.
	 *
	 * @param timestampAndWatermarkAssigner The implementation of the timestamp assigner and
	 *                                      watermark generator.
	 * @return The stream after the transformation, with assigned timestamps and watermarks.
	 * @see AssignerWithPeriodicWatermarks
	 * @see AssignerWithPunctuatedWatermarks
	 * @see #assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks)
	 */
	public SingleOutputMixStreamOperator<T> assignTimestampsAndWatermarks(
		AssignerWithPeriodicWatermarks<T> timestampAndWatermarkAssigner) {

		// match parallelism to input, otherwise dop=1 sources could lead to some strange
		// behaviour: the watermark will creep along very slowly because the elements
		// from the source go to each extraction operator round robin.
		final int inputParallelism = getTransformation().getParallelism();
		final AssignerWithPeriodicWatermarks<T> cleanedAssigner = clean(timestampAndWatermarkAssigner);

		TimestampsAndPeriodicWatermarksOperator<T> operator =
			new TimestampsAndPeriodicWatermarksOperator<>(cleanedAssigner);

		return transform("Timestamps/Watermarks", getTransformation().getOutputType(), operator)
			.setParallelism(inputParallelism);
	}

	/**
	 * Assigns timestamps to the elements in the data stream and creates watermarks to
	 * signal event time progress based on the elements themselves.
	 *
	 * <p>This method creates watermarks based purely on stream elements. For each element
	 * that is handled via {@link AssignerWithPunctuatedWatermarks#extractTimestamp(Object, long)},
	 * the {@link AssignerWithPunctuatedWatermarks#checkAndGetNextWatermark(Object, long)}
	 * method is called, and a new watermark is emitted, if the returned watermark value is
	 * non-negative and greater than the previous watermark.
	 *
	 * <p>This method is useful when the data stream embeds watermark elements, or certain elements
	 * carry a marker that can be used to determine the current event time watermark.
	 * This operation gives the programmer full control over the watermark generation. Users
	 * should be aware that too aggressive watermark generation (i.e., generating hundreds of
	 * watermarks every second) can cost some performance.
	 *
	 * <p>For cases where watermarks should be created in a regular fashion, for example
	 * every x milliseconds, use the {@link AssignerWithPeriodicWatermarks}.
	 *
	 * @param timestampAndWatermarkAssigner The implementation of the timestamp assigner and
	 *                                      watermark generator.
	 * @return The stream after the transformation, with assigned timestamps and watermarks.
	 * @see AssignerWithPunctuatedWatermarks
	 * @see AssignerWithPeriodicWatermarks
	 * @see #assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks)
	 */
	public SingleOutputMixStreamOperator<T> assignTimestampsAndWatermarks(
		AssignerWithPunctuatedWatermarks<T> timestampAndWatermarkAssigner) {

		// match parallelism to input, otherwise dop=1 sources could lead to some strange
		// behaviour: the watermark will creep along very slowly because the elements
		// from the source go to each extraction operator round robin.
		final int inputParallelism = getTransformation().getParallelism();
		final AssignerWithPunctuatedWatermarks<T> cleanedAssigner = clean(timestampAndWatermarkAssigner);

		TimestampsAndPunctuatedWatermarksOperator<T> operator =
			new TimestampsAndPunctuatedWatermarksOperator<>(cleanedAssigner);

		return transform("Timestamps/Watermarks", getTransformation().getOutputType(), operator)
			.setParallelism(inputParallelism);
	}

	// ------------------------------------------------------------------------
	//  Data sinks
	// ------------------------------------------------------------------------

	/**
	 * Writes a DataMixStream to the standard output stream (stdout).
	 *
	 * <p>For each element of the DataMixStream the result of {@link Object#toString()} is written.
	 *
	 * <p>NOTE: This will print to stdout on the machine where the code is executed, i.e. the Gaia
	 * worker.
	 *
	 * @return The closed DataMixStream.
	 */
	@PublicEvolving
	public DataMixStreamSink<T> print() {
		PrintSinkFunction<T> printFunction = new PrintSinkFunction<>();
		return addSink(printFunction).name("Print to Std. Out");
	}

	/**
	 * Writes a DataMixStream to the standard output stream (stderr).
	 *
	 * <p>For each element of the DataMixStream the result of {@link Object#toString()} is written.
	 *
	 * <p>NOTE: This will print to stderr on the machine where the code is executed, i.e. the Gaia
	 * worker.
	 *
	 * @return The closed DataMixStream.
	 */
	@PublicEvolving
	public DataMixStreamSink<T> printToErr() {
		PrintSinkFunction<T> printFunction = new PrintSinkFunction<>(true);
		return addSink(printFunction).name("Print to Std. Err");
	}

	/**
	 * Writes a DataMixStream to the standard output stream (stdout).
	 *
	 * <p>For each element of the DataMixStream the result of {@link Object#toString()} is written.
	 *
	 * <p>NOTE: This will print to stdout on the machine where the code is executed, i.e. the Gaia
	 * worker.
	 *
	 * @param sinkIdentifier The string to prefix the output with.
	 * @return The closed DataMixStream.
	 */
	@PublicEvolving
	public DataMixStreamSink<T> print(String sinkIdentifier) {
		PrintSinkFunction<T> printFunction = new PrintSinkFunction<>(sinkIdentifier, false);
		return addSink(printFunction).name("Print to Std. Out");
	}

	/**
	 * Writes a DataMixStream to the standard output stream (stderr).
	 *
	 * <p>For each element of the DataMixStream the result of {@link Object#toString()} is written.
	 *
	 * <p>NOTE: This will print to stderr on the machine where the code is executed, i.e. the Gaia
	 * worker.
	 *
	 * @param sinkIdentifier The string to prefix the output with.
	 * @return The closed DataMixStream.
	 */
	@PublicEvolving
	public DataMixStreamSink<T> printToErr(String sinkIdentifier) {
		PrintSinkFunction<T> printFunction = new PrintSinkFunction<>(sinkIdentifier, true);
		return addSink(printFunction).name("Print to Std. Err");
	}

	/**
	 * Writes a DataMixStream to the file specified by path in text format.
	 *
	 * <p>For every element of the DataMixStream the result of {@link Object#toString()} is written.
	 *
	 * @param path The path pointing to the location the text file is written to.
	 * @return The closed DataMixStream.
	 */
	@PublicEvolving
	public DataMixStreamSink<T> writeAsText(String path) {
		return writeUsingOutputFormat(new TextOutputFormat<T>(new Path(path)));
	}


	/**
	 * Writes a DataMixStream to the file specified by path in text format.
	 *
	 * <p>For every element of the DataMixStream the result of {@link Object#toString()} is written.
	 *
	 * @param path      The path pointing to the location the text file is written to
	 * @param writeMode Controls the behavior for existing files. Options are
	 *                  NO_OVERWRITE and OVERWRITE.
	 * @return The closed DataMixStream.
	 */
	@PublicEvolving
	public DataMixStreamSink<T> writeAsText(String path, WriteMode writeMode) {
		TextOutputFormat<T> tof = new TextOutputFormat<>(new Path(path));
		tof.setWriteMode(writeMode);
		return writeUsingOutputFormat(tof);
	}


	/**
	 * Writes a DataMixStream to the file specified by the path parameter.
	 *
	 * <p>For every field of an element of the DataMixStream the result of {@link Object#toString()}
	 * is written. This method can only be used on data streams of tuples.
	 *
	 * @param path the path pointing to the location the text file is written to
	 * @return the closed DataMixStream
	 */
	@PublicEvolving
	public DataMixStreamSink<T> writeAsCsv(String path) {
		return writeAsCsv(path, null, CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
	}


	/**
	 * Writes a DataMixStream to the file specified by the path parameter.
	 *
	 * <p>For every field of an element of the DataMixStream the result of {@link Object#toString()}
	 * is written. This method can only be used on data streams of tuples.
	 *
	 * @param path      the path pointing to the location the text file is written to
	 * @param writeMode Controls the behavior for existing files. Options are
	 *                  NO_OVERWRITE and OVERWRITE.
	 * @return the closed DataMixStream
	 */
	@PublicEvolving
	public DataMixStreamSink<T> writeAsCsv(String path, WriteMode writeMode) {
		return writeAsCsv(path, writeMode, CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
	}

	/**
	 * Writes a DataMixStream to the file specified by the path parameter. The
	 * writing is performed periodically every millis milliseconds.
	 *
	 * <p>For every field of an element of the DataMixStream the result of {@link Object#toString()}
	 * is written. This method can only be used on data streams of tuples.
	 *
	 * @param path           the path pointing to the location the text file is written to
	 * @param writeMode      Controls the behavior for existing files. Options are
	 *                       NO_OVERWRITE and OVERWRITE.
	 * @param rowDelimiter   the delimiter for two rows
	 * @param fieldDelimiter the delimiter for two fields
	 * @return the closed DataMixStream
	 */
	@SuppressWarnings("unchecked")
	@PublicEvolving
	public <X extends Tuple> DataMixStreamSink<T> writeAsCsv(
		String path,
		WriteMode writeMode,
		String rowDelimiter,
		String fieldDelimiter) {
		Preconditions.checkArgument(
			getType().isTupleType(),
			"The writeAsCsv() method can only be used on data streams of tuples.");

		CsvOutputFormat<X> of = new CsvOutputFormat<>(
			new Path(path),
			rowDelimiter,
			fieldDelimiter);

		if (writeMode != null) {
			of.setWriteMode(writeMode);
		}

		return writeUsingOutputFormat((OutputFormat<T>) of);
	}

	/**
	 * Writes the DataMixStream to a socket as a byte array. The format of the
	 * output is specified by a {@link SerializationSchema}.
	 *
	 * @param hostName host of the socket
	 * @param port     port of the socket
	 * @param schema   schema for serialization
	 * @return the closed DataMixStream
	 */
	@PublicEvolving
	public DataMixStreamSink<T> writeToSocket(String hostName, int port, SerializationSchema<T> schema) {
		DataMixStreamSink<T> returnMixStream = addSink(new SocketClientSink<>(hostName, port, schema, 0));
		returnMixStream.setParallelism(1); // It would not work if multiple instances would connect to the same port
		return returnMixStream;
	}

	/**
	 * Writes the dataMixStream into an output, described by an OutputFormat.
	 *
	 * <p>The output is not participating in Gaia's checkpointing!
	 *
	 * <p>For writing to a file system periodically, the use of the "gaia-connector-filesystem"
	 * is recommended.
	 *
	 * @param format The output format
	 * @return The closed DataMixStream
	 */
	@PublicEvolving
	public DataMixStreamSink<T> writeUsingOutputFormat(OutputFormat<T> format) {
		return addSink(new OutputFormatSinkFunction<>(format));
	}

	/**
	 * Method for passing user defined operators along with the type
	 * information that will transform the DataMixStream.
	 *
	 * @param operatorName name of the operator, for logging purposes
	 * @param outTypeInfo  the output type of the operator
	 * @param operator     the object containing the transformation logic
	 * @param <R>          type of the return stream
	 * @return the data stream constructed
	 */
	@PublicEvolving
	public <R> SingleOutputMixStreamOperator<R> transform(String operatorName, TypeInformation<R> outTypeInfo, OneInputMixStreamOperator<T, R> operator) {

		// read the output type of the input Transform to coax out errors about MissingTypeInfo
		transformation.getOutputType();

		OneInputTransformation<T, R> resultTransform = new OneInputTransformation<>(
			this.transformation,
			operatorName,
			operator,
			outTypeInfo,
			environment.getParallelism());

		@SuppressWarnings({"unchecked", "rawtypes"})
		SingleOutputMixStreamOperator<R> returnMixStream = new SingleOutputMixStreamOperator(environment, resultTransform);

		getExecutionEnvironment().addOperator(resultTransform);

		return returnMixStream;
	}

	/**
	 * Internal function for setting the partitioner for the DataMixStream.
	 *
	 * @param partitioner Partitioner to set.
	 * @return The modified DataMixStream.
	 */
	protected DataMixStream<T> setConnectionType(MixStreamPartitioner<T> partitioner) {
		return new DataMixStream<>(this.getExecutionEnvironment(), new PartitionTransformation<>(this.getTransformation(), partitioner));
	}

	/**
	 * Adds the given sink to this DataMixStream. Only streams with sinks added
	 * will be executed once the {@link MixStreamExecutionEnvironment#execute()}
	 * method is called.
	 *
	 * @param sinkFunction The object containing the sink's invoke function.
	 * @return The closed DataMixStream.
	 */
	public DataMixStreamSink<T> addSink(SinkFunction<T> sinkFunction) {

		// read the output type of the input Transform to coax out errors about MissingTypeInfo
		transformation.getOutputType();

		// configure the type if needed
		if (sinkFunction instanceof InputTypeConfigurable) {
			((InputTypeConfigurable) sinkFunction).setInputType(getType(), getExecutionConfig());
		}

		MixStreamSink<T> sinkOperator = new MixStreamSink<>(clean(sinkFunction));

		DataMixStreamSink<T> sink = new DataMixStreamSink<>(this, sinkOperator);

		getExecutionEnvironment().addOperator(sink.getTransformation());
		return sink;
	}

	/**
	 * Returns the {@link MixStreamTransformation} that represents the operation that logically creates
	 * this {@link DataMixStream}.
	 *
	 * @return The Transformation
	 */
	@Internal
	public MixStreamTransformation<T> getTransformation() {
		return transformation;
	}
}